package org.example.flink.connections.pulsar;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;

import java.util.concurrent.TimeUnit;

public class TestDelive {

    private static boolean flag = true;

    public static void main(String[] args) throws Exception{

        PulsarClient client = PulsarClient
                .builder().enableTransaction(true)
                .serviceUrl("http://127.0.0.1:8080").build();

        Producer<String> producer = client.newProducer(Schema.STRING)
                .topic("persistent://public/default/delay10")
                .producerName("test")
                .sendTimeout(0, TimeUnit.SECONDS)
                .create();

        String line = null;
        int cnt = 0;
        while (flag){
            line = String.valueOf(cnt)+" date="+ DateUtil.date();
            producer.newMessage()
                    .key(String.valueOf(cnt))
//                    .deliverAfter(5,TimeUnit.SECONDS)
                    .value(line)
                    .send();

            System.out.println(line);
            Thread.sleep(1000);
            cnt++;
            if(cnt > 2000){
                flag = false;
            }
        }
        producer.close();
        client.close();
    }


}
